package com.smsc.headend.task.engine.dao.impl;

import cn.hutool.core.date.SystemClock;
import cn.hutool.json.JSONUtil;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.data.dto.TaskDataDTO;
import com.smsc.headend.module.data.dto.TaskPacketDTO;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.BaseTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.entity.DeviceTask;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.module.task.dto.AtomicTaskState;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.assertj.core.util.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Repository
@Slf4j
public class TaskDAOImpl implements TaskDAO {
    @Autowired
    RedisUtils redisUtils;

    Cache<String, AtomicTaskState> atomicTaskStateCache = CacheBuilder.newBuilder().expireAfterWrite(300, TimeUnit.MINUTES).build();


    @Override
    public void terminalTaskQueueAdd(Task task) {
        if (task == null || task.getTaskNo() == null) {
            return;
        }
        String key = getTerminalTaskListKey(task);
        BaseTask baseTask = new BaseTask();
        baseTask.setTaskNo(task.getTaskNo());
        baseTask.setTaskType(task.getTaskType());
        String taskTerminalKey = RedisKeys.getTaskTerminalKey(key);
        Long currentLen = redisUtils.llen(taskTerminalKey);
        log.debug("com {} queue size {}, add taskNo={}", task.getComId(), currentLen, task.getTaskNo());
        redisUtils.lpush(taskTerminalKey, JSONUtil.toJsonStr(baseTask));
        redisUtils.set(RedisKeys.getTaskKey(task.getTaskNo()), JSONUtil.toJsonStr(task), 7 * 24 * 60 * 60);
    }

    /**
     * get first task in Task Queue
     *
     * @param terminalQueueId
     * @return
     */
    @Override
    public Task terminalTaskQueueGet(String terminalQueueId) {
        String taskJson = redisUtils.lGetIndexString(RedisKeys.getTaskTerminalKey(terminalQueueId), -1l);
        if (StringUtils.isEmpty(taskJson)) {
            return null;
        }
        BaseTask baseTask = JSONUtil.toBean(taskJson, BaseTask.class);
        Task task = this.getTaskByNo(baseTask.getTaskNo());
        if (task == null) {
            redisUtils.rpop(RedisKeys.getTaskTerminalKey(terminalQueueId));
            return terminalTaskQueueGet(terminalQueueId);
        }
        return task;
    }


    /**
     * get task by index in Task Queue
     *
     * @param terminalQueueId
     * @param index
     * @return
     */
    @Override
    public Task terminalTaskQueueGetByIndex(String terminalQueueId, Integer index) {
        String taskJson = redisUtils.lGetIndexString(RedisKeys.getTaskTerminalKey(terminalQueueId), (-1 * index) - 1);
        if (StringUtils.isEmpty(taskJson)) {
            return null;
        }
        BaseTask baseTask = JSONUtil.toBean(taskJson, BaseTask.class);
        return this.getTaskByNo(baseTask.getTaskNo());
    }

    @Override
    public Long getTerminalTaskLength(String terminalId) {
        String taskTerminalKey = RedisKeys.getTaskTerminalKey(terminalId);
        return redisUtils.llen(taskTerminalKey);
    }

    @Override
    public Task getTaskByNo(String taskNo) {
        String taskContentJson = (String) redisUtils.get(RedisKeys.getTaskKey(taskNo));
        Task task = JSONUtil.toBean(taskContentJson, Task.class);
        if (task != null && StringUtils.isEmpty(task.getTaskNo())) {
            return null;
        }
        return task;
    }

    @Override
    public boolean updateTask(Task task) {
        return redisUtils.set(RedisKeys.getTaskKey(task.getTaskNo()), JSONUtil.toJsonStr(task), 7 * 24 * 60 * 60);
    }

    /**
     * pop first task in Task Queue
     *
     * @param terminalQueueId
     */
    @Override
    public void terminalTaskQueuePop(String terminalQueueId) {
        String taskJson = redisUtils.rpop(RedisKeys.getTaskTerminalKey(terminalQueueId));
        if (StringUtils.isEmpty(taskJson)) {
            return;
        }
        BaseTask baseTask = JSONUtil.toBean(taskJson, BaseTask.class);
        Task task = getTaskByNo(baseTask.getTaskNo());
        if (task != null) {
            redisUtils.del(RedisKeys.getTaskKey(baseTask.getTaskNo()));
        }
        log.debug("task execute: task pop. terminalQueueId={},taskNo={}", terminalQueueId, baseTask.getTaskNo());
    }


    @Override
    public AtomicTask getProcessingAtomicTask(String taskNo) {
        Object object = redisUtils.get(RedisKeys.getProcessingTaskKey(taskNo));
        if (object != null) {
            return JSONUtil.toBean(String.valueOf(object), AtomicTask.class);
        }
        return null;
    }


    @Override
    public Set<String> hashGetAtomicTaskStateSet() {
        return redisUtils.hKeys(RedisKeys.getProcessingAtomicTaskKey());
    }

    @Override
    public boolean hashSetProcessingTerminalQueue(String termId, Integer priority) {
        if (termId == null || priority == null) {
            log.error("termId {} priority {} error", termId, priority);
            return false;
        }
        return redisUtils.hset(RedisKeys.getProcessingTerminal(), termId, priority);
    }

    @Override
    public Integer hashGetProcessingTerminalQueue(String termId) {
        Object terminalKey = redisUtils.hget(RedisKeys.getProcessingTerminal(), termId);
        if (terminalKey == null) {
            return null;
        }
        return Integer.valueOf(String.valueOf(terminalKey));
    }

    @Override
    public List<String> hashKeyProcessingTerminalQueue() {
        return Lists.newArrayList(redisUtils.hKeys(RedisKeys.getProcessingTerminal()));
    }

    @Override
    public void hashDelProcessingTerminalQueue(String terminalId) {
        redisUtils.hdel(RedisKeys.getProcessingTerminal(), terminalId);
    }

    @Override
    public AtomicTask atomicTaskQueueGet(String taskNo) {
        String taskKey = RedisKeys.getProcessingTaskKey(taskNo);
        return JSONUtil.toBean(redisUtils.lGetIndexString(taskKey, -1l), AtomicTask.class);
    }

    @Override
    public void atomicTaskQueueAdd(AtomicTask atomicTask) {
        redisUtils.lpush(RedisKeys.getProcessingTaskKey(atomicTask.getTaskNo()), JSONUtil.toJsonStr(atomicTask));
        redisUtils.expire(RedisKeys.getProcessingTaskKey(atomicTask.getTaskNo()), 7 * 24 * 60 * 60);
    }

    @Override
    public void atomicTaskQueueAddReverse(AtomicTask atomicTask) {
        redisUtils.rpush(RedisKeys.getProcessingTaskKey(atomicTask.getTaskNo()), JSONUtil.toJsonStr(atomicTask));
    }

    @Override
    public AtomicTask atomicTaskQueuePopReverse(String taskNo) {
        String atomicTaskJson = redisUtils.lpop(RedisKeys.getProcessingTaskKey(taskNo));
        if (StringUtils.isNotEmpty(atomicTaskJson)) {
            return JSONUtil.toBean(atomicTaskJson, AtomicTask.class);
        }
        return null;
    }

    @Override
    public AtomicTask atomicTaskQueuePop(String taskNo) {
        String atomicTaskJson = redisUtils.rpop(RedisKeys.getProcessingTaskKey(taskNo));
        if (StringUtils.isNotEmpty(atomicTaskJson)) {
            AtomicTask at = JSONUtil.toBean(atomicTaskJson, AtomicTask.class);
            log.debug("pop out taskNO={},atomicTask={}", taskNo, at.getAtomicTaskNo());
            return at;
        }
        return null;
    }

    @Override
    public void atomicTaskQueueDelete(String taskNo) {
        log.debug("delete task {} atomic queue", taskNo);
        redisUtils.del(RedisKeys.getProcessingTaskKey(taskNo));
    }

    @Override
    public boolean atomicTaskQueueExist(String taskNo) {
        return redisUtils.hasKey(RedisKeys.getProcessingTaskKey(taskNo));
    }

    @Override
    public void hashDeleteAtomicTaskState(String atomicTaskNo, Integer priority) {
        log.debug("hdel {} {}", RedisKeys.getProcessingAtomicTaskKey(), atomicTaskNo);
        atomicTaskStateCache.invalidate(atomicTaskNo);
        if (priority <= Task.Priority.ABOVENORMAL.getPriorityCode()) {
            redisUtils.del(RedisKeys.getAtomicTaskMapTask(atomicTaskNo));
        }
    }

    @Override
    public void hashSetAtomicTaskState(String atomicTaskNo, Task task, int sendTimes, AtomicTaskState state) {
        if (state != null && state.isStored()) {
            return;
        }
        Long executionTime = SystemClock.now() / 1000;
        AtomicTaskState st = new AtomicTaskState();
        st.setAtomicTaskNo(atomicTaskNo);
        st.setTerminalQueueId(getTerminalTaskListKey(task));
        st.setLastTime(executionTime);
        st.setRetryTimes(sendTimes);
        st.setTaskNo(task.getTaskNo());
        if (state != null) {
            st.setHandlerClass(state.getHandlerClass());
            st.setBlockNumber(state.getBlockNumber());
            st.setCollectingEndTv(state.getCollectingEndTv());
            st.setCollectedEndTv(state.getCollectedEndTv());
            st.setData(state.getData());
            st.setMatchedBlockNumbers(state.getMatchedBlockNumbers());
        }
        atomicTaskStateCache.put(atomicTaskNo, st);
//        if (task.getPriority() <= Task.Priority.ABOVENORMAL.getPriorityCode()) {
        redisUtils.set(RedisKeys.getAtomicTaskMapTask(atomicTaskNo), JSONUtil.toJsonStr(state), 24 * 60 * 60);
//        }
    }

    @Override
    public AtomicTaskState hashGetAtomicTaskState(String atomicTaskNo) {
        AtomicTaskState stateLocal = atomicTaskStateCache.getIfPresent(atomicTaskNo);
        return stateLocal == null ? hashGetAtomicTaskStatePublicCache(atomicTaskNo) : stateLocal;
    }

    private AtomicTaskState hashGetAtomicTaskStatePublicCache(String atomicTaskNo) {
        String jsonStr = (String) redisUtils.get(RedisKeys.getAtomicTaskMapTask(atomicTaskNo));
        if (StringUtils.isEmpty(jsonStr)) {
            return null;
        }
        AtomicTaskState state = JSONUtil.toBean(jsonStr, AtomicTaskState.class);
        return state;
    }

    @Override
    public String getTerminalTaskListKey(Task task) {
        return getTerminalTaskListKey(task.getComId(), task.getPriority());
    }

    @Override
    public String getTerminalTaskListKey(Long terminalId, Integer priority) {
        return String.format("%s:%s", terminalId, priority);
    }

    @Override
    public List<TaskDataDTO> getAllTaskData(String taskNo) {
        String key = RedisKeys.getPopTaskDataKey(taskNo);
        List<Object> allData = redisUtils.lRange(key, 0, -1);
        if (CollectionUtils.isEmpty(allData)) {
            return null;
        }
        return allData.stream().map(o ->
                JSONUtil.toBean(String.valueOf(o), TaskDataDTO.class)).collect(Collectors.toList());
    }

    @Override
    public void addTaskData(String taskNo, String atomicTaskNo, String paramName, Object dataObject, Long expireSeconds, Boolean isSuccess) {
        String key = RedisKeys.getPopTaskDataKey(taskNo);
        TaskDataDTO dataDTO = new TaskDataDTO();
        dataDTO.setAtomicTaskNo(atomicTaskNo);
        dataDTO.setKey(paramName);
        dataDTO.setValue(dataObject);
        dataDTO.setIsSuccess(isSuccess);
        redisUtils.lpush(key, JSONUtil.toJsonStr(dataDTO));
        if (expireSeconds != null && expireSeconds > 0) {
            redisUtils.expire(key, expireSeconds);
        }
    }

    @Override
    public List<TaskDataDTO> popAllTaskData(String taskNo) {
        String key = RedisKeys.getPopTaskDataKey(taskNo);

        Object data;
        List<TaskDataDTO> result = new ArrayList<>();
        while ((data = redisUtils.rpop(key)) != null) {
            TaskDataDTO map = JSONUtil.toBean(String.valueOf(data), TaskDataDTO.class);
            result.add(map);
        }
        return result;
    }

    @Override
    public List<TaskPacketDTO> popAllTaskPacket(String taskNo) {
        String key = RedisKeys.getTaskPacketKey(taskNo);
        Object data;
        List<TaskPacketDTO> result = new ArrayList<>();
        while ((data = redisUtils.rpop(key)) != null) {
            TaskPacketDTO map = JSONUtil.toBean(String.valueOf(data), TaskPacketDTO.class);
            result.add(map);
        }
        return result;
    }

    @Override
    public List<TaskPacketDTO> popAllTaskPacket(DeviceTask deviceTask) {
        String key = RedisKeys.getTaskPacketKey(deviceTask.getTaskNo());
        Object data;
        List<TaskPacketDTO> result = new ArrayList<>();
        while ((data = redisUtils.rpop(key)) != null) {
            TaskPacketDTO map = JSONUtil.toBean(String.valueOf(data), TaskPacketDTO.class);
            if (null == map.getMeterId()) {
                map.setMeterId(deviceTask.getDeviceId());
            }
            result.add(map);
        }
        return result;
    }


    @Override
    public void setTaskProcessingAtomicTaskNo(String taskNo, String atomicTaskNo) {
        redisUtils.set(RedisKeys.getTaskProcessingKey(taskNo), atomicTaskNo, 60 * 60);
    }

    @Override
    public void clearTaskProcessingAtomic(String taskNo) {
        redisUtils.del(RedisKeys.getTaskProcessingKey(taskNo));
    }

    @Override
    public String getProcessingAtomicTaskByTaskNo(String taskNo) {
        Object atomicTaskNo = redisUtils.get(RedisKeys.getTaskProcessingKey(taskNo));
        return atomicTaskNo == null ? null : String.valueOf(atomicTaskNo);
    }

    @Override
    public void hashSetRetryTask(Task task) {
        redisUtils.hset(RedisKeys.getRetryTaskKey(), task.getTaskNo(), JSONUtil.toJsonStr(task));
    }

}
